package org.eden.rpc.core.client.pooled;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.DefaultEventExecutor;
import io.netty.util.concurrent.EventExecutor;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.Promise;
import lombok.extern.slf4j.Slf4j;
import org.eden.rpc.common.entity.RpcRequest;
import org.eden.rpc.core.client.NettyClient;
import org.eden.rpc.core.client.NettyClientHandler;
import org.eden.rpc.core.client.RpcClient;
import org.eden.rpc.core.codec.DefaultDecoder;
import org.eden.rpc.core.codec.DefaultEncoder;
import org.eden.rpc.core.codec.KryoSerializer;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
public class SimpleBlockingNettyConnectionPool implements NettyConnectionPool {

    private static final int DEFAULT_CAPACITY = 1 << 8;
    private static final int WAITING_TIME_SECONDS = 10;

    // default is zero
    private final AtomicInteger clientNum;
    private final int maxPoolSize;

    private final String host;
    private final int port;

    private final BlockingQueue<RpcClient> pool;

    /**
     * DefaultEventExecutor 继承了 SingleThreadEventExecutor，该 executor 持有了一个任务队列，所以可以同时将多个创建任务交给 executor,
     * executor 负责将溢出的请求加入队列
     * notifier 负责唤醒 Future 的持有线程
     */
    private final EventExecutor executor = new DefaultEventExecutor();
    private final EventExecutor notifier = new DefaultEventExecutor();

    public SimpleBlockingNettyConnectionPool(String host, int port) {
        this(host, port, DEFAULT_CAPACITY);
    }

    public SimpleBlockingNettyConnectionPool(String host, int port, int capacity) {
        this.pool = new LinkedBlockingDeque<>(capacity);
        this.host = host;
        this.port = port;
        this.maxPoolSize = capacity;
        this.clientNum = new AtomicInteger(0);
    }

    public void retrieve(RpcClient client) {
        if (!healthCheck(client)) {
            log.debug("client-" + client + ": 失去连接");
            clientNum.decrementAndGet();
            return;
        }
        if (!pool.offer(client)) {
            clientNum.decrementAndGet();
            log.debug("丢弃多余 netty 连接对象");
        }
    }

    public Future<RpcClient> getConnection() {
        Promise<RpcClient> promise = notifier.newPromise();
        return doGetConnection(promise);
    }

    private Future<RpcClient> doGetConnection(Promise<RpcClient> promise) {
        try {
            RpcClient client = pool.poll(0, TimeUnit.MILLISECONDS);
            if (client != null) {
                promise.trySuccess(client);
                return promise;
            }
        } catch (InterruptedException e) {
            log.error(e.getMessage());
            promise.tryFailure(e);
            return promise;
        }
        if (clientNum.get() < maxPoolSize) {
            if (clientNum.incrementAndGet() <= maxPoolSize) {
                createNewConnection(promise);
            } else {
                clientNum.decrementAndGet();
                return doGetConnection(promise);
            }
        } else {
            // 需要等待
            executor.execute(() -> {
                try {
                    RpcClient client = pool.poll(WAITING_TIME_SECONDS, TimeUnit.SECONDS);
                    promise.trySuccess(client);
                } catch (InterruptedException e) {
                    log.error(e.getMessage());
                    promise.tryFailure(e);
                }
            });
        }
        return promise;
    }

    private void createNewConnection(Promise<RpcClient> promise) {
        Bootstrap bootstrap = newBootstrap();
        NettyClient client = new NettyClient(bootstrap);
        ChannelFuture future = bootstrap.connect();
        if (future.isDone()) {
            initConnection(client, future, promise);
        } else {
            future.addListener((future1) -> initConnection(client, future, promise));
        }
    }

    public void initConnection(NettyClient client, ChannelFuture future, Promise<RpcClient> promise) {
        if (future.isSuccess()) {
            client.setChannel(future.channel());
            client.setHandler(future.channel().pipeline().get(NettyClientHandler.class));
            promise.trySuccess(client);
        } else {
            promise.tryFailure(future.cause());
        }
    }

    public boolean healthCheck(Connection client) {
        return client.isValid();
    }

    private Bootstrap newBootstrap() {
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .remoteAddress(host, port)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        pipeline.addLast(new DefaultEncoder<>(new KryoSerializer(), RpcRequest.class));
                        pipeline.addLast(new DefaultDecoder());
                        pipeline.addLast(new NettyClientHandler());
                    }
                });
        return bootstrap;
    }
}
